package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class Demo8Index {

    private HConnection connection;

    private HBaseAdmin hBaseAdmin;

    @Before
    public void init() {
        // 1、创建连接

        Configuration configuration = new Configuration();

        /*
         * 不需要执行hmaster和regionserver的地址m  hclient会自动去zookeeper中获取，
         * 由于获取到的是一个域名，所以本地必须配置hosts
         *
         */

        //设置zookeeper地址
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        try {

            //创建连接  对数据增删改查使用
            connection = HConnectionManager.createConnection(configuration);

            /*
             * 创建修改删除表使用
             *
             */
            hBaseAdmin = new HBaseAdmin(connection);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 数据表的没有数据索引表中也没有数据的时候
     */

    @Test
    public void putDataAndIndex() throws IOException {


        /**
         * 先创建索引表和数据表
         * create_namespace 'bigdata'
         * create 'bigdata:student','info'
         * create 'bigdata:student_index_gender','info'
         */

        HTableInterface student = connection.getTable(TableName.valueOf("bigdata:student"));
        HTableInterface index = connection.getTable(TableName.valueOf("bigdata:student_index_gender"));


        BufferedReader bufferedReader = new BufferedReader(new FileReader("data/students.txt"));

        String line;
        while ((line = bufferedReader.readLine()) != null) {
            String[] split = line.split(",");
            String id = split[0];
            String name = split[1];
            String age = split[2];
            String gender = split[3];
            String clazz = split[4];


            ArrayList<Put> puts = new ArrayList<>();

            ArrayList<Put> puts1 = new ArrayList<>();

            //增加数据量让学号表的不唯一
            for (int i = 0; i < 1000; i++) {

                //学号表的不唯一
                String newid = id + i;

                Put put = new Put(newid.getBytes());

                put.add("info".getBytes(), "name".getBytes(), name.getBytes());
                put.add("info".getBytes(), "age".getBytes(), age.getBytes());
                put.add("info".getBytes(), "gender".getBytes(), gender.getBytes());
                put.add("info".getBytes(), "clazz".getBytes(), clazz.getBytes());

                //存入一条数据
                puts.add(put);

                Put put1 = new Put(gender.getBytes());

                put1.add("info".getBytes(), newid.getBytes(), null);

                //保存索引   以性别作为rowkey,以学号作为列名  列值为空
                puts1.add(put1);
//                index.put(put1);
            }

            student.put(puts);
            index.put(puts1);

        }

    }


    /**
     * 查询数据
     * <p>
     * 先查索引表再查数据表
     */
    @Test
    public void getIndexAndData() throws Exception {


        long start = System.currentTimeMillis();


        /**
         *
         * 查询性别为男的学生
         *
         */

        String gender = "男";


        HTableInterface student = connection.getTable(TableName.valueOf("bigdata:student"));
        HTableInterface index = connection.getTable(TableName.valueOf("bigdata:student_index_gender"));


        //哈寻索引表
        Get get = new Get(gender.getBytes());
        Result result = index.get(get);

        //返回所有列
        List<Cell> cells = result.listCells();


        //要查询的学生
        ArrayList<Get> gets = new ArrayList<>(50000);


        for (Cell cell : cells) {
            //取出列名  列名就是学号
            String id = Bytes.toString(CellUtil.cloneQualifier(cell));


            Get get1 = new Get(id.getBytes());
            gets.add(get1);

        }


        //查询数据表   批量查询    由于新别为难的学生太多, 导致查询效率低
        Result[] results = student.get(gets);

        for (Result result1 : results) {
            String name = Bytes.toString(result1.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(result1.getValue("info".getBytes(), "age".getBytes()));
            String gender1 = Bytes.toString(result1.getValue("info".getBytes(), "gender".getBytes()));
            String clazz = Bytes.toString(result1.getValue("info".getBytes(), "clazz".getBytes()));


            //System.out.println(name + "," + age + "," + gender1 + "," + clazz);
        }


        long end = System.currentTimeMillis();

        System.out.println(end - start);

    }

    /**
     * 不适用索引表
     */

    @Test
    public void scanFilter() throws Exception {

        long start = System.currentTimeMillis();

        String gender = "男";

        HTableInterface student = connection.getTable(TableName.valueOf("bigdata:student"));


        Scan scan = new Scan();

        //二进制比较器
        BinaryComparator binaryComparator = new BinaryComparator(gender.getBytes());

        //列值过滤器
        SingleColumnValueFilter filter = new SingleColumnValueFilter("info".getBytes(),
                "gender".getBytes(), CompareFilter.CompareOp.EQUAL, binaryComparator);


        scan.setFilter(filter);


        //全表扫描
        ResultScanner scanner = student.getScanner(scan);


        //打印结果
        print(scanner);


        long end = System.currentTimeMillis();

        System.out.println(end - start);


    }


    public static void print(ResultScanner scanner) throws IOException {
        Result next;

        while ((next = scanner.next()) != null) {

            String gender = Bytes.toString(next.getValue("info".getBytes(), "gender".getBytes()));
            String name = Bytes.toString(next.getValue("info".getBytes(), "name".getBytes()));
            String age = Bytes.toString(next.getValue("info".getBytes(), "age".getBytes()));
            String clazz = Bytes.toString(next.getValue("info".getBytes(), "clazz".getBytes()));
            String id = Bytes.toString(next.getRow());

            //System.out.println(id + "\t" + name + "\t" + age + '\t' + gender + '\t' + clazz);
        }
    }

}
